#include "ace_udp_acceptor_impl.h"

//using namespace vpm::mina;
using namespace vpm::mina::detail;

static ACE_THR_FUNC_RETURN udp_acceptor_reactor_event_loop (void *pArg)
{
	ACE_Reactor *pReactor = ACE_static_cast (ACE_Reactor *, pArg);
	pReactor->owner(ACE_OS::thr_self());
	pReactor->run_reactor_event_loop ();
	return 0;
}


int AceUdpAcceptorImpl::bind(int port, const std::string & host/*="0.0.0.0"*/)
{
	this->reactor(m_pReactor);
	m_thrdGrpId = ACE_Thread_Manager::instance()->spawn_n(1 , udp_acceptor_reactor_event_loop,m_pReactor);

	if (m_bRecvQueue)
	{
		this->activate();
	}

	ACE_INET_Addr listenAddr(port,host.c_str());

	int ret = m_dgramSock.open(listenAddr);
	ACE_DEBUG((LM_DEBUG,ACE_TEXT("start datagram network : %s:%d\n"), listenAddr.get_host_addr(),listenAddr.get_port_number())); 
	if (ret == -1)
	{			
		ACE_DEBUG((LM_DEBUG,ACE_TEXT("start datagram network : %s:%d failed\n"), listenAddr.get_host_addr(),listenAddr.get_port_number())); 
		exit(-1); 
	}else 
	{
		if (this->reactor()->register_handler(this, ACE_Event_Handler::READ_MASK) == -1)
		{
			ACE_ERROR_RETURN ((LM_ERROR,"%p", "Failed"),-1);
		}			
	}
	return 0; 
}

void AceUdpAcceptorImpl::unbind()
{
	if (m_bRecvQueue)
	{
		ACE_Message_Block * pStopMsg = new ACE_Message_Block(0,ACE_Message_Block::MB_STOP);
		this->putq(pStopMsg); 
		this->wait(); 
	}

	m_dgramSock.close(); 

	if (m_thrdGrpId != -1 )
	{
		m_pReactor->remove_handler(this,ACE_Event_Handler::READ_MASK); 
		m_pReactor->end_reactor_event_loop(); 
		ACE_Thread_Manager::instance()->wait_grp(m_thrdGrpId);
	}
}


int AceUdpAcceptorImpl::handle_input(ACE_HANDLE handle)
{	
	ACE_INET_Addr fromAddr;

	iovec ioData ; 
	ioData.iov_base = NULL; 
	ioData.iov_len = 0; 
	//size_t recvLen = this->m_dgramSock.recv (recvBuf,DGRAM_NET_MAX_RECV_BUF_LEN, fromAddr);
	this->m_dgramSock.recv(&ioData, fromAddr); 

	if (ioData.iov_len ==0  )
	{
		return 0; 
	}

	if (ioData.iov_len > UDP_MAX_RECV_BUF_LEN )
	{
		delete [] (char*)ioData.iov_base; 
		//drop the very big packet 
		return 0; 
	}

	if (m_bRecvQueue)
	{	
		ACE_Message_Block * pMBox  = new ACE_Message_Block (sizeof(ACE_INET_Addr)); 

		pMBox->copy((char*)(&fromAddr),sizeof(ACE_INET_Addr));

		ACE_Message_Block * pCont = new ACE_Message_Block ((char *)ioData.iov_base, 0);       
		pCont->wr_ptr((int)ioData.iov_len); 

		pMBox->cont(pCont); 

		ACE_Time_Value wait(0,100);
		wait += ACE_OS::gettimeofday(); 
		if (this->msg_queue()->is_full() || this->putq(pMBox,&wait) == -1) 
		{			
			ACE_DEBUG((LM_ERROR,ACE_TEXT("Receive message queue is full")));			
			pMBox->release(); 
			delete [] (char*)ioData.iov_base; 
		}
	}
	else 
	{
		this->process_data((char*)ioData.iov_base,ioData.iov_len,fromAddr.get_port_number(), fromAddr.get_host_addr()); 
	}
	return 0; 
}

int AceUdpAcceptorImpl::svc() 
{
	ACE_Message_Block * pMBox; 
	ACE_INET_Addr *pRemoteAddr ; 
	while(true) 
	{
		int ret = this->getq(pMBox);
		if (ret >=0 )   
		{
			if (pMBox->msg_type() == ACE_Message_Block::MB_STOP)
			{
				pMBox->release(); 
				ACE_DEBUG((LM_DEBUG,ACE_TEXT("Stopping network task ...")));				
				return 0; 
			}
			pRemoteAddr = ACE_reinterpret_cast(ACE_INET_Addr*, pMBox->rd_ptr());

			pMBox->rd_ptr(ACE_static_cast (size_t, pMBox->length()));
			ACE_Message_Block * pCont = pMBox->cont();
			char * pMsg = pCont->rd_ptr();
			int msgLen = (int)pCont->length();

			this->process_data(pMsg,msgLen,pRemoteAddr->get_port_number(),pRemoteAddr->get_host_addr()); 
			pMBox->release(); 
			delete[] pMsg; 
			pMsg = NULL;
		}
	}

	ACE_DEBUG((LM_DEBUG,ACE_TEXT("Fatel error, reactor work task exit!!!")));
	return 0; 
}
